package com.beta.mq.redis.producer.impl;

import com.alibaba.fastjson.JSON;
import com.beta.mq.redis.RedisMqProperties;
import com.beta.mq.redis.model.Message;
import com.beta.mq.redis.producer.Producer;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;

/**
 * @author taofucheng
 * @date 2015/12/8.
 */
public class RedisBlockingQueueProducer implements Producer {
    private RedisMqProperties properties;
    private RedissonClient redissonClient;

    public RedisBlockingQueueProducer(RedissonClient redissonClient, RedisMqProperties properties) {
        this.redissonClient = redissonClient;
        this.properties = properties;
    }

    @Override
    public void sendMessage(String topic, Message message) throws InterruptedException {
        // 先判断size是不是已经超限制了
        RBlockingQueue<String> queue = redissonClient.getBlockingQueue(topic);
        if (queue != null && queue.size() > properties.getQueueSize()) {
            //超过限制后，丢弃老的，将新的插入进去
            queue.take();
        }
        queue.put(JSON.toJSONString(message));
    }

}
